package com.kingjw.sourceTest

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

/**
 *
 * @Package: com.kingjw.sourceTest
 * @ClassName: JDBCSink
 * @Author: 王坚伟
 * @CreateTime: 2022/1/17 19:21
 * @Description:
 */
object JDBCSink {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 这里需要设置为1，不然会因为线程问题导致update失败
    env.setParallelism(1)
    val inputStream: DataStream[String] = env.readTextFile("input/sensor.txt")
    // 先转换成样例类类型（简单转换操作）
    val dataStream= inputStream
      .map( data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).trim.toLong, arr(2).trim.toDouble)
      } )
    dataStream.print()
    val value: DataStreamSink[SensorReading] = dataStream.addSink(new MyJdbcSink)
    env.execute("jdbc sink test")
  }
}

case class SensorReading(id: String, timestamp: Long, temperature: Double)

class MyJdbcSink() extends RichSinkFunction[SensorReading]{

  var conn: Connection = _
  var insertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/mybatis", "root", "123456")
    insertStmt = conn.prepareStatement("insert into temperatures(sensor,temp) values(?,?)")
    updateStmt = conn.prepareStatement("update temperatures set temp = ? where sensor = ?")
  }

  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    updateStmt.setDouble(1,value.temperature)
    updateStmt.setString(2,value.id)
    updateStmt.execute()

    if (updateStmt.getUpdateCount == 0){
      insertStmt.setString(1,value.id)
      insertStmt.setDouble(2,value.temperature)
      insertStmt.execute()
    }

  }

  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }
}